package com.xc.hotelmgr.consumermq.config;

import com.alibaba.fastjson.JSONObject;
import com.xc.hotelmgr.consumermq.pojo.mq.*;
import com.xc.hotelmgr.consumermq.pojo.mybatis.HtDailyroom;
import com.xc.hotelmgr.consumermq.pojo.mybatis.HtHotelinfo;
import com.xc.hotelmgr.consumermq.pojo.mybatis.HtLogicRoom;
import com.xc.hotelmgr.consumermq.service.FeignService.CtripFeginService;
import com.xc.hotelmgr.consumermq.service.FeignService.MtFeginService;
import com.xc.hotelmgr.consumermq.service.mq.MQService;
import com.xc.hotelmgr.consumermq.utils.SendMessageMgr;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;


@Slf4j
@Configuration
public class ConsumerConfig {

    @Value("${RabbitMQ.virtual-host}")
    private String virtualHost;

    @Value("${RabbitMQ.host}")
    private String host;

    @Value("${RabbitMQ.port}")
    private String port;

    @Value("${RabbitMQ.username}")
    private String username;

    @Value("${RabbitMQ.password}")
    private String password;

    @Autowired
    private MQService mqService;

    @Autowired
    private MtFeginService mtFeginService;

    @Autowired
    private CtripFeginService ctripFeginService;
    /**
     * 创建连接工厂
     * @return
     */
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(host + ":" +port);
        connectionFactory.setVirtualHost(virtualHost);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        return connectionFactory;
    }

    @Bean
    public SendMessageMgr rabbitMessage(){
        return  new SendMessageMgr();
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        //用RabbitAdmin一定要配置这个，spring加载的是后就会加载这个类================特别重要
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }
    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        return rabbitTemplate;
    }

    @Bean("sendAll")
    public DirectMessageListenerContainer directMessageListenerContainerSendAll(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setConsumersPerQueue(3); //  最大的消费者数量
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
//        container.setQueueNames("sendAll_"+ appKey);
//        container.setQueueNames("sendAll_24670810");
        container.setPrefetchCount(1);
        container.setMessageListener((MessageListener) message -> {
            String queueName = message.getMessageProperties().getConsumerQueue();
            byte[] msg = message.getBody();
            SendAll sendAll = JSONObject.parseObject(msg, SendAll.class);
            try {
                mqService.hotelSendAll(sendAll, queueName);
            } catch (Exception e) {
                log.error(e.toString());
            }
        });
        return container;
    }

    /**
     * 监听 "htUpDown_" + appKey 消息队列，上下架或删除酒店
     * @param connectionFactory
     * @return
     */
    @Bean("upDown")
    public DirectMessageListenerContainer directMessageListenerContainerHtUpDown(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
//        container.setConsumersPerQueue(5); //  最大的消费者数量
//        container.setQueueNames("htUpDown_" + appKey);
        container.setPrefetchCount(1);
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        container.setMessageListener((MessageListener) message -> {
            String queueName = message.getMessageProperties().getConsumerQueue();
            byte[] msg = message.getBody();
            UpDown upDown = JSONObject.parseObject(msg, UpDown.class);
            try {
                mqService.hotelUpDown(upDown, queueName);
            } catch (Exception e) {
                log.error(e.toString());
            }
        });
        return container;
    }

    /**
     * 监听hotel_id_list消息队列，如果是拉取分组酒店并推送，
     *      * 调用datapusher-ali-onece 的 /hotel/send_all 推送价格计划及房态
     * @param connectionFactory
     * @return
     */
    @Bean("htPull")
    public DirectMessageListenerContainer directMessageListenerContainerHtPull(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
//        container.setConsumersPerQueue(1); //  最大的消费者数量
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        container.setPrefetchCount(1);
        container.setQueueNames("hotel_id_list");
        container.setMessageListener((MessageListener) message -> {
            byte[] msg = message.getBody();
            TaskHotelList taskHotelList = JSONObject.parseObject(msg, TaskHotelList.class);
            try {
                //携程酒店id已-1开头
                if(taskHotelList.getHotelList().size() !=0 && taskHotelList.getHotelList().get(0).toString().substring(0,2).equals("-1")){
                    mqService.ctripSendAll(taskHotelList);
                }else {
                    mqService.pullHotel(taskHotelList);
                }

            } catch (Exception e) {
                log.error(e.toString());
            }
        });

        return container;
    }

    @Bean("htCallback")
    public DirectMessageListenerContainer directMessageListenerContainerHtCallback(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
//        container.setConsumersPerQueue(1); //  最大的消费者数量
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
//        container.setQueueNames("callback_hotel_32393251");
        container.setPrefetchCount(1);
        container.setMessageListener((MessageListener) message -> {
            String queueName = message.getMessageProperties().getConsumerQueue();
            byte[] msg = message.getBody();
            HotelCallBack hotelCallBack = JSONObject.parseObject(msg, HotelCallBack.class);
            try {
                mqService.hotelCallBack(hotelCallBack, queueName);
            } catch (Exception e) {
                log.error(e.toString());
            }

        });
        return container;
    }

    @Bean("rpCallback")
    public DirectMessageListenerContainer directMessageListenerContainerRpCallback(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
//        container.setConsumersPerQueue(2); //  最大的消费者数量
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
//        container.setQueueNames("callback_rp_" + appKey);
        container.setPrefetchCount(1);
        container.setMessageListener((MessageListener) message -> {
            String queueName = message.getMessageProperties().getConsumerQueue();
            byte[] msg = message.getBody();
            RpCallBack rpCallBack = JSONObject.parseObject(msg, RpCallBack.class);
            try {
                mqService.rpCallBack(rpCallBack, queueName);
            } catch (Exception e) {
                log.error(e.toString());
            }
        });
        return container;
    }

    @Bean("ratesCallback")
    public DirectMessageListenerContainer directMessageListenerContainerRatesCallback(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
//        container.setQueueNames("callback_rates_32565667");
//        container.setConsumersPerQueue(1);
        container.setPrefetchCount(1);
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        container.setMessageListener((MessageListener) message -> {
            String queueName = message.getMessageProperties().getConsumerQueue();
            byte[] msg = message.getBody();
            RatesCallBack ratesCallBack = JSONObject.parseObject(msg, RatesCallBack.class);
            try {
                mqService.ratesCallBack(ratesCallBack, queueName);
            } catch (Exception e) {
                log.error("callback_rates出错，队列名称：{}，rp信息：{}，异常信息：{}",queueName,ratesCallBack,e.toString());
            }
        });
        return container;
    }

    @Bean("MQ2MQHotel")
    public DirectMessageListenerContainer directMessageListenerContainerMQ2MQHotel(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
//        container.setConsumersPerQueue(1); //  最大的消费者数量
        container.setPrefetchCount(1);
        container.setQueueNames("change_hotel");
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        container.setMessageListener((MessageListener) message -> {
            byte[] msg = message.getBody();

            //1、查询酒店所属店铺信息及酒店下历史rp信息
            //2、删除数据（酒店、房型、rp、rps)
            //3、拉取（酒店、房型、rp、rps)
            //4、删除redis(无效rp所属的店铺、酒店、房型信息)
            //5、保存到redis(rp所使用的店铺、酒店、房型以及rps）
            //6、计算失效出失效的rp，将店铺信息、酒店信息、失效rp信息发送到call_back消息队列
            try {
                HtHotelinfo htHotelinfo = JSONObject.parseObject(msg, HtHotelinfo.class);
                mqService.dealAndSend(htHotelinfo);
            }catch (Exception ex){
                log.error("MQ2MQHotel:{}", msg);
            }

        });
        return container;
    }

    @Bean("MQ2MQRp")
    public DirectMessageListenerContainer directMessageListenerContainerMQ2MQRp(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
//        container.setConsumersPerQueue(5); //  最大的消费者数量
        container.setQueueNames("change_rp");
        container.setPrefetchCount(1);
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        container.setMessageListener((MessageListener) message -> {
            byte[] msg = message.getBody();
            List<HtLogicRoom> htLogicRooms = JSONObject.parseObject(msg, List.class);
            try {
                mtFeginService.saveRp2MQ(htLogicRooms);
            } catch (Exception e) {
                log.error(e.toString());
            }

        });
        return container;
    }

    /**
     * 监听变化的房态
     * @param connectionFactory
     * @return
     */
    @Bean("MQ2MQRates")
    public DirectMessageListenerContainer directMessageListenerContainerMQ2MQRates(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
//        container.setConsumersPerQueue(200); //  最大的消费者数量
        container.setQueueNames("change_rates");
        container.setPrefetchCount(1);
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        container.setMessageListener((MessageListener) message -> {
            byte[] msg = message.getBody();
            List<HtDailyroom> htDailyrooms = JSONObject.parseObject(msg, List.class);
            try {
                if (!htDailyrooms.isEmpty()){

                    String t =  JSONObject.toJSONString(htDailyrooms.get(0));
                    HtDailyroom htDailyroom = JSONObject.parseObject(t,HtDailyroom.class);
                    Long rpCode = htDailyroom.getRpCode();
                    if (rpCode<0){
                        ctripFeginService.saveRates2MQ(htDailyrooms);
                    }else {
                        mtFeginService.saveRates2MQ(htDailyrooms);
                    }
                }

            } catch (Exception e) {
                log.error(e.toString());
            }
        });
        return container;
    }

    /**
     * 删除价格计划
     * @param connectionFactory
     * @return
     */
    @Bean("DelRp")
    public DirectMessageListenerContainer directMessageListenerContainerDelRp(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
//        container.setConsumersPerQueue(5); //  最大的消费者数量
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        container.setPrefetchCount(1);
//        container.setQueueNames("delete_rp_" + appKey);
        container.setMessageListener((MessageListener) message -> {
            String queueName = message.getMessageProperties().getConsumerQueue();
            byte[] msg = message.getBody();
            DelRpMQ delRpMQ = JSONObject.parseObject(msg, DelRpMQ.class);
            try {
                mqService.rpDel(delRpMQ, queueName);
            } catch (Exception e) {
                log.error(e.toString());
            }
        });
        return container;
    }

    /**
     * 从 消息队列 hotel_id_list_all（暂存所有酒店ID），转移消息 到 hotel_id_list（实际拉取队列）队列进行拉取
     * 从前端hotelIdListAll(on_off/{status})可以暂停（移除对 hotel_id_list_all 队列的监听，转移）
     * @param connectionFactory
     * @return
     */
    @Bean("HotelIdListAll")
    public DirectMessageListenerContainer directMessageListenerContainerDelRpHotelIdListAll(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
//        container.setConsumersPerQueue(5); //  最大的消费者数量
        container.setQueueNames("hotel_id_list_all");
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        container.setPrefetchCount(1);
        container.setMessageListener((MessageListener) message -> {
            byte[] msg = message.getBody();
            TaskHotelList taskHotelList = JSONObject.parseObject(msg, TaskHotelList.class);
            try {
                mqService.moveHtId(taskHotelList);
            } catch (InterruptedException e) {
                log.error(e.toString());
            }

        });
        return container;
    }

    @Bean("CMQ2MQRates")
    public DirectMessageListenerContainer directMessageListenerContainerUpHtId(ConnectionFactory connectionFactory){
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
//        container.setConsumersPerQueue(5); //  最大的消费者数量
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        container.setQueueNames("xc_change_rates");
        container.setPrefetchCount(1);
        container.setMessageListener((MessageListener) message -> {
//            String queueName = message.getMessageProperties().getConsumerQueue();
            byte[] msg = message.getBody();
            JSONObject jsonObject = JSONObject.parseObject(msg, JSONObject.class);
            try { ;
                mqService.upHtId(jsonObject);
                Thread.sleep(60);
            } catch (Exception e) {
                log.error(e.toString());
            }
        });
        return container;
    }

}